package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/hpcloud/tail"
	"github.com/robfig/cron"
	"go.etcd.io/etcd/clientv3"
	"log"
	"logstash/config"
	"logstash/logger"
	"logstash/models"
	"logstash/mq"
	"logstash/mq/queue"
	"logstash/utile"
	"strings"
	"time"
)

const (
	StatusNormal = 1  // 正常状态
	StatusDelete = -1 // 删除状态
)

//读取日志

func readLog(t *utile.TailObj) {
	log.Printf("Watching log file [ %s ] ...\n", t.FileName)
	date := time.Now().Format("20060102")
	fileName := fmt.Sprintf("%s/%s-%s.%s", t.FilePath, t.FileName, date, t.Ext)
	if t.Tail == nil {
		tf, err := tail.TailFile(fileName, tail.Config{
			ReOpen:    true,
			Follow:    true,
			Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
			MustExist: false,
			Poll:      true,
		})
		if err != nil {
			return
		}
		t.Tail = tf
	}
	fmt.Printf("--------------- before filename: %s -------------\n", t.Tail.Filename)
	//日志按天生成，需比对文件名，来读取最新的日志文件
	if strings.Compare(t.Tail.Filename, fileName) != 0 {
		tf, err := tail.TailFile(fileName, tail.Config{
			ReOpen:    true,
			Follow:    true,
			Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
			MustExist: false,
			Poll:      true,
		})
		if err != nil {
			return
		}
		t.Tail = tf
	}
	fmt.Printf("--------------- after filename: %s -------------\n", t.Tail.Filename)
	//读取每行日志内容
	for line := range t.Tail.Lines {
		if line.Err != nil {
			logger.Errorf("read line failed,err:%v", line.Err)
			continue
		}
		str := strings.TrimSpace(line.Text)
		if len(str) == 0 || str[0] == '\n' {
			fmt.Println("------------- no read------------")
			continue
		}
		fmt.Printf("-------------------------------读取到日志 [ %s ] [ %s ]：%s\n", t.Tail.Filename, t.Id, str)
		var msg = models.MessageBag{
			MsgType: t.FileName,
			Content: str,
		}
		SendLog2Mq(queue.PRODUCER_TEST, msg)

	}
}

//发送日志到MQ
func SendLog2Mq(queueName string, msg models.MessageBag) {
	msgByt, _ := json.Marshal(msg)
	defer func() {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			fmt.Printf("SendLog2Mq 休息3秒,err:%s\n", err)
			SendLog2Mq(queueName, msg)
		}
	}()
	producer, ok := mq.GetProducer(queueName)
	if ok {

		err := producer.Send(string(msgByt))
		if err == nil {
			fmt.Println("发送日志到MQ成功")
		} else {
			logger.Errorf("发送日志到MQ失败，err：%v", err)
		}
	}
}

func watchConfig() {
	etcd := utile.GetEtcd()
	ip := utile.GetOutboundIP()
	key := fmt.Sprintf("/%s/%s", etcd.GetPrefix(), ip)
	// 初始化连接etcd
	cli, err := utile.GetEtcdClient()
	if err != nil {
		logger.Error("连接etcd失败:", err)
		return
	}
	logger.Info("开始监控key:", key)

	conf := config.GetConfig()

	// Watch操作
	for {
		var getConfSucc = true
		wch := cli.Watch(context.Background(), key)
		for resp := range wch {
			for _, ev := range resp.Events {
				// DELETE处理
				if ev.Type == clientv3.EventTypeDelete {
					logger.Warnf("删除Key[%s]配置", key)
					continue
				}
				// PUT处理
				if ev.Type == clientv3.EventTypePut && string(ev.Kv.Key) == key {
					err = json.Unmarshal(ev.Kv.Value, conf.Data)
					if err != nil {
						logger.Error("反序列化key[%s]失败:", err)
						getConfSucc = false
						continue
					}
				}
				logger.Debugf("get config from etcd ,Type: %v, Key:%v, Value:%v\n", ev.Type, string(ev.Kv.Key), string(ev.Kv.Value))
			}
			if getConfSucc {
				logger.Debugf("get config from etcd success, %v", conf.Data)
				//TODO 通知文件监控
				//_ = tailf.UpdateConfig(collectConf)
				updateConfig()
			}
		}
	}
}

func updateConfig() {
	fmt.Println("监控日志变化 UpdateConfig")
	conf := config.GetConfig()
	tmgr := utile.GetTailMgr()
	for _, logfile := range conf.Data.LogFiles {
		var isNew = true
		//配置变动文件被放弃监控，从map中剔除，并close对应的监控goroutine
		for filename, _ := range tmgr.TailObjMap {
			if filename == logfile {
				isNew = false
				break
			}
		}
		if isNew {
			fmt.Println("add watching new log file: " + logfile)
			tmgr.AddLogFile(logfile)
			go readLog(tmgr.TailObjMap[logfile])
		}
	}
	var status int
	for filename, tailobj := range tmgr.TailObjMap {
		status = StatusDelete
		for _, logfile := range conf.Data.LogFiles {
			if filename == logfile {
				status = StatusNormal
				break
			}
		}
		//配置变动文件被放弃监控，从map中剔除，并close对应的监控goroutine
		if status == StatusDelete {
			fmt.Println("release watching log file: " + filename)
			delete(tmgr.TailObjMap, filename)
			tailobj.Tail.Stop()
			continue
		}
	}
}

//监控日志文件
func watchLogFiles() {
	//启动文件监控
	tailMgr := utile.NewTailMgr()
	conf := config.GetConfig()
	for _, fileName := range conf.Data.LogFiles {
		err := tailMgr.AddLogFile(fileName)
		if err != nil {
			logger.Error("add log file failed,err:%v", err)
			continue
		}
	}
	fmt.Printf("-------------进入监控协程：%v\n", tailMgr.TailObjMap)
	for _, tailObj := range tailMgr.TailObjMap {
		go readLog(tailObj)
	}
}

//关闭日志文件监控
func closeWatchLogFiles() {
	time.Sleep(time.Second * 1)
	//启动文件监控
	tailMgr := utile.GetTailMgr()
	fmt.Printf("-------------进入关闭协程：%v\n", tailMgr.TailObjMap)
	for _, tailObj := range tailMgr.TailObjMap {
		fmt.Println("关闭协程：" + tailObj.Id)
		tailObj.Tail.Stop()
	}
}

//每天更改监控的文件（日志文件都是按日期命名生成）
func changeLogFileDayByDay() {
	c := cron.New()
	spec := "0 0 0 * * ?" //每日0晨切换新日志
	//spec := "*/30 * * * * ?"
	err := c.AddFunc(spec, func() {
		logger.Infof("cron running: %s", time.Now().Format("2006-01-02 15:04:05"))
		closeWatchLogFiles()
		watchLogFiles()
	})
	if err != nil {
		fmt.Printf("cron err: %s\n", err)
		logger.Errorf("cron err: %s\n", err)
	}
	c.Start()

}

//程序入口
func Run() {

	go watchConfig()

	watchLogFiles()

	changeLogFileDayByDay()

	fmt.Println("程序启动SUCC")
	//MQ发送测试数据
	//for i := 1; i < 10; i++ {
	//	SendLog2Mq(queue.PRODUCER_TEST, fmt.Sprintf("this is my mq test %v ", i))
	//}
	//time.Sleep(time.Second * 3)
	//for i := 1; i < 10; i++ {
	//	SendLog2Mq(queue.PRODUCER_USER, fmt.Sprintf("user_id: %v ", i))
	//}

}
